# 实时分析资金流入情况，找到资金入流的股票排行，注意不是流入资金大小，而是流入的趋势；（求资金流入的斜率）
import pandas as pd
import os
import sys
import time
import numpy as np
import datetime
import threading
from clickhouse_driver import Client

lock = threading.Lock()

# 资金流入数据ETL，包括合并处理，导数ck数据库等；正式使用 20230817


#全局
mylist = []
# 多线程并发执行
def process_zjlr_merge_all_concurrency(path, save_to):
    if os.path.exists(save_to):
        print("文件已经存在，跳过！")
        return

    start = time.time()
    pd.set_option('display.unicode.ambiguous_as_wide', True)  # 处理数据的列标题与数据无法对齐的情况
    pd.set_option('display.unicode.east_asian_width', True)  # 无法对齐主要是因为列标题是中文
    pd.set_option('display.max_columns', None)

    list = os.listdir(path)
    for i in range(0, len(list)):
        # 指定字段类型，否则000删除了
        if os.path.isfile(os.path.join(path, list[i])):
            # 多线程执行
            t = threading.Thread(name='t_' + list[i], target=merge_csv, args=(os.path.join(path, list[i]),))
            t.start()
            t.join()  # 阻塞，直到执行完毕，才会执行主线程
        else:
            print("跳过文件夹：%s" % (list[i]))

    print("开始合并...")
    # print(mylist)
    df = pd.DataFrame(mylist, columns={'采集日期': str, '股票代码': str, '时间': str, '主力净流入': int, '小单净流入': int, '中单净流入': int,
                                       '大单净流入': int, '超大单净流入': int})
    print(df.head())
    # 需要等多线程执行完毕
    print("save...")
    df["股票代码"] = df["股票代码"].astype(str)

    # 增加一列，从时间列读取date数据
    # 先在正确位置插入一列，再赋值
    df.insert(2, "trade_date", "", False)
    df["trade_date"] = df["时间"].str.slice(0, 10)
    print(df.head())
    # 保存的时候000被删除了,RE:文件中没有删除000，但是excel打开时看不到000，
    df.to_csv(save_to, encoding="utf_8_sig", index=False)  # index设置为False，就不会有第一列的序号
    end = time.time()
    print("concurrency time: %s" % (end - start))

def merge_csv(path):
    append = pd.read_csv(path, dtype={'股票代码': str}, encoding="UTF-8",
                         index_col=False)  # 取消索引列，否则错位
    print("线程：%s 分析文件： %s" % (threading.current_thread().name, path))
    size = len(append)
    append = append.iloc[0:size, :]  # 行，列
    # 需要加锁
    lock.acquire()
    for item in append.values.tolist():
        mylist.append(item)
    lock.release()


#离线批处理，单个文件（合并后的文件）
def batch_process(path):
    client = Client(host="192.168.50.40", user="default", password="9defbcg", port="9090", database="ods")
    print("start batch process ZJLR data...")
    #读取csv数据，必须指定类型，否则会推测为int，导致异常
    df = pd.read_csv(path, dtype={'采集日期': str, '股票代码': str,'trade_date': str, '主力净流入':np.int64, '小单净流入':np.int64, '中单净流入':np.int64, '大单净流入':np.int64, '超大单净流入':np.int64}
                     , encoding="UTF-8", index_col=False)  # 取消索引列，否则错位
    client.execute('INSERT INTO ods_stock_zjlr_df VALUES', df.values.tolist())
    #关闭连接 TODO
    print("batch process stock ZJLR data finished!")

# 必须有main方法，调度才能执行
def main():
    print("zjrl_etl start....")
    path = sys.argv[1]
    save_to = sys.argv[2]
    # 合并资金流入数据 耗时：30s
    process_zjlr_merge_all_concurrency(path, save_to)
    #批量导入到CK
    batch_process(save_to)


if __name__ == '__main__':
    main()



